package com.atguigu.day02;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//Filter 语义：将流中每个元素转换成0个或者1个元素
public class Example5 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //匿名内部类方式
        env
                .addSource(new ClickSource())
                .filter(new FilterFunction<ClickEvent>() {
                    @Override
                    public boolean filter(ClickEvent value) throws Exception {
                        return value.username.equals("Jordan");
                    }
                })
                .print();

        //外部类方式
        env
                .addSource(new ClickSource())
                .filter(new MyFilter())
                .print();

        //匿名函数方式
        env
                .addSource(new ClickSource())
                .filter(r -> r.username.equals("Allen"))
                .print();

        //使用flatMap实现filter
       env
               .addSource(new ClickSource())
               .flatMap(new FlatMapFunction<ClickEvent, ClickEvent>() {
                   @Override
                   public void flatMap(ClickEvent value, Collector<ClickEvent> out) throws Exception {
                       if(value.username.equals("Allen")) out.collect(value);
                   }
               })
               .print();



        env.execute();
    }

    public static class MyFilter implements FilterFunction<ClickEvent>{
        @Override
        public boolean filter(ClickEvent value) throws Exception {
            return value.username.equals("Iverson");
        }
    }


}
